package cn.jly.flink.state;

import cn.jly.flink.entity.SensorReading;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 同一个传感器连续两次温度差超过10度进行报警
 * 同时测试了keyedProcessFunction
 *
 * @PackageName cn.jly.flink.state
 * @ClassName StateBackendDemo
 * @Description 状态后端和容错demo
 * @Author 姬岚洋
 * @Date 2021/2/4 下午 1:54
 */
public class StateBackendAndFaultToleranceDemo {

    private static final String HIGH_TEMP_DIFF_TAG = "high_temp_diff";
    private StreamExecutionEnvironment env;

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();

        // 1. 设置状态后端
        // env.setStateBackend(new MemoryStateBackend());
        // env.setStateBackend(new FsStateBackend("hdfs://node01:9000/xxx"));
        env.setStateBackend(new RocksDBStateBackend("file:///d:/test/flink/state/"));

        // 2. 检查点配置
        // env.enableCheckpointing(500, CheckpointingMode.EXACTLY_ONCE);
        env.enableCheckpointing(500);

        // 高级设置
        env.getCheckpointConfig().setCheckpointingMode(CheckpointConfig.DEFAULT_MODE);
        // 最好是要设置一个超时时间，这边设置1分钟
        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
        // 多个checkpoint任务同时进行（分布式环境下，可能一个checkpoint任务还没结束，下一个checkpoint已经开始了，flink允许这样）
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        // 两个checkpoint任务之间的间隔，表示上一个checkpoint任务的结束到下一个checkpoint任务的开始不能小于设置的100毫秒
        // 这么做的意义是可以留出点时间来进行数据处理计算，因为一个计算器的计算资源总是有限的，即使不同线程但也需要共享cpu
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100L);
        // 设置允许checkpoint失败多少次（超过次数则整个任务失败，数据也无法继续处理）
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);

        // 3. 重启策略的配置
        // ① 固定延迟重启：每隔1分钟尝试重启，尝试三次
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.minutes(1)));
        // ②失败率重启：10分钟内尝试3次重启，每次重启间隔1分钟
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10), Time.minutes(1)));

        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        final String host = parameterTool.get("host", "node01");
        final int port = parameterTool.getInt("port", 9999);

        // 侧输出流outputTag
        final OutputTag<Tuple3<String, SensorReading, SensorReading>> highTempDiffTag = new OutputTag<>(
                HIGH_TEMP_DIFF_TAG,
                TypeInformation.of(new TypeHint<Tuple3<String, SensorReading, SensorReading>>() {
                })
        );

        final SingleOutputStreamOperator<SensorReading> processDatStream = env.socketTextStream(host, port)
                .flatMap(new FlatMapFunction<String, SensorReading>() {
                    @Override
                    public void flatMap(String value, Collector<SensorReading> out) throws Exception {
                        if (StringUtils.isNotEmpty(value)) {
                            final String[] split = value.split(",");
                            out.collect(new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2])));
                        }
                    }
                })
                .keyBy(new KeySelector<SensorReading, String>() {
                    @Override
                    public String getKey(SensorReading value) throws Exception {
                        return value.getId();
                    }
                })
                .process(new KeyedProcessFunction<String, SensorReading, SensorReading>() {
                    // 状态存储上一个传感器信息
                    transient ValueState<SensorReading> lastSensorReadingValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);

                        final ValueStateDescriptor<SensorReading> stateDescriptor = new ValueStateDescriptor<>(
                                "上一个传感器信息",
                                TypeInformation.of(new TypeHint<SensorReading>() {
                                })
                        );
                        lastSensorReadingValueState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
                        // 取出状态值
                        final SensorReading lastSensorReading = lastSensorReadingValueState.value();
                        if (lastSensorReading != null) {
                            final double tempDiff = Math.abs(value.getTemperature() - lastSensorReading.getTemperature());
                            if (tempDiff > 10d) {
                                ctx.output(highTempDiffTag, Tuple3.of("两次温差超过10度", lastSensorReading, value));
                            }
                        }
                        // 更新状态值
                        lastSensorReadingValueState.update(value);

//                        final Long timestamp = ctx.timestamp();
//                        final long l = ctx.timerService().currentWatermark();
//                        final long l = ctx.timerService().currentProcessingTime();
                        // 注册处理时间定时器，当前处理时间2秒后触发
                        ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 10L);

                        // 输出当前传感器信息
                        out.collect(value);
                    }

                    /**
                     * 定时器触发
                     *
                     * @param timestamp
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SensorReading> out) throws Exception {
                        System.out.println("StateBackendAndFaultToleranceDemo.onTimer 定时器触发了 ~~~~~" + timestamp);
                    }
                });

        // 正常信息
        processDatStream.print("当前");

        // 高温警报
        processDatStream.getSideOutput(highTempDiffTag).printToErr("警报");

        FlinkUtils.executeStream(env, "StateBackendAndFaultToleranceDemo");
    }
}
